package my.storm.zmq;

import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;
import org.zeromq.ZMQException;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;

public class ZmqSpout implements IRichSpout {

	/**
	 * 
	 */
	private static final long serialVersionUID = -320694160373381094L;
	
	private Logger LOG = LoggerFactory.getLogger(this.getClass());
	
	private SpoutOutputCollector collector;
	private ZMQ.Context zmqContext;
	private ZMQ.Socket zmqSocket;

	@SuppressWarnings("rawtypes")
	public void open(Map conf, TopologyContext context,
			SpoutOutputCollector collector) {
		
		this.collector = collector;
		
		this.zmqContext = ZMQ.context(1);
		this.zmqSocket = this.zmqContext.socket(ZMQ.PULL);
		this.zmqSocket.connect("tcp://211.113.22.153:5556");
		
		LOG.info(">>>>> zmqContext={}, zmqSocket={}", this.zmqContext, this.zmqSocket);
		LOG.info(">>>>> ZmqSpout({}) opended", Thread.currentThread().getId());
	}

	public void close() {

		if(zmqSocket != null) {
			zmqSocket.close();
			zmqSocket = null;
		}
		
		if(zmqContext != null) {
			zmqContext.term();
			zmqContext = null;
		}
	}

	public void nextTuple() {

		Utils.sleep(100);
		LOG.debug(">>>>> nextTuple zmqSocket={}", zmqSocket);
		
		if(this.zmqSocket != null) {
			try {
//				byte[] byteReceived = null;
				byte[] byteReceived = this.zmqSocket.recv(0);
				
				LOG.info(">>> byteReceived={}", byteReceived);
				this.collector.emit(new Values(byteReceived));
			} catch(ZMQException e) {
				throw new RuntimeException(e.getCause());
			}
		}
	}

	public void ack(Object msgId) {
		// TODO Auto-generated method stub
		
	}

	public void fail(Object msgId) {
		// TODO Auto-generated method stub
		
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {

		declarer.declare(new Fields("byteReceived"));
	}

	public boolean isDistributed() {
		// TODO Auto-generated method stub
		return false;
	}

}
